package com.smsc.headend.task.engine.handler.impl.generalparam;

import cn.hutool.core.convert.Convert;
import cn.hutool.core.util.StrUtil;
import cn.hutool.json.JSONObject;
import cn.hutool.json.JSONUtil;
import com.smsc.headend.common.enums.errorcode.TaskHandlerErrorCode;
import com.smsc.headend.common.utils.RedisUtils;
import com.smsc.headend.module.data.enums.OperationGroup;
import com.smsc.headend.module.protocol.DeviceData;
import com.smsc.headend.module.scheme.entity.ConfigScheduleTask;
import com.smsc.headend.module.task.consts.TaskKafkaTopic;
import com.smsc.headend.module.task.dto.AtomicTask;
import com.smsc.headend.module.task.dto.Task;
import com.smsc.headend.module.task.enums.DeviceActionType;
import com.smsc.headend.module.task.enums.SubTaskStatus;
import com.smsc.headend.module.task.enums.TaskType;
import com.smsc.headend.task.engine.dao.TaskDAO;
import com.smsc.headend.task.engine.exception.UDISTaskHandlerException;
import com.smsc.headend.task.engine.handler.TaskHandler;
import com.smsc.headend.task.engine.service.FeignAssetManagementService;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;

import java.nio.charset.StandardCharsets;
import java.util.Map;

/**
 * @author wxh
 * @date 2021/4/20 19:23
 */
public class SetGeneralParamHandler implements TaskHandler {
    @Autowired
    TaskDAO taskDAO;
    @Autowired
    KafkaTemplate kafkaTemplate;
    @Autowired
    RedisUtils redisUtils;
    @Autowired
    FeignAssetManagementService assetManagementRemoteService;
    @Override
    public int handle(Task task, AtomicTask atomicTask, DeviceData cosemData, String actionResult) throws UDISTaskHandlerException {
        log.info("{} action result {}", atomicTask.getAtomicTaskNo(), actionResult);
        Map<String, Object> param = getParameterMap(task, atomicTask);
        Long subDataItemId = Long.valueOf(String.valueOf(param.get("subDataItemId"))) ;
        if (param.get("dataChannel") != null && param.get("dataChannel").equals("true")) {
            // 数据通道报文下发 可能有get set action, 执行成功的话都返回sent success,不做解析处理
            actionResult = TaskHandlerErrorCode.PACKET_SENT_SUCCESS.getErrorText();
        }
        if (task.getIsODR()) {
            taskDAO.addTaskData(task.getTaskNo(), atomicTask.getAtomicTaskNo(), Convert.toStr(subDataItemId), actionResult, 600l, Boolean.TRUE);
        }
        // Vending接口下发结果处理
        if (StrUtil.equals(TaskType.WsMdmSetParameter.toString(), task.getTaskType())) {
            handleSetParameterResult(task, atomicTask, true, "");
        }
        if (StrUtil.equalsIgnoreCase(task.getTaskType(),TaskType.SetRelayControlMode.name())) {
            assetManagementRemoteService.updateLogRelayControl(task.getTaskNo(), task.getMeterId(), ConfigScheduleTask.ActionType.SET_DEVICE_CONFIG.getType(), SubTaskStatus.Success.getCode(),null, null);
        }
        return 1;
    }

    @Override
    public int handleFailure(Task task, AtomicTask atomicTask, String message) {
        log.info("error {}, ", message);
        Map<String, Object> param = getParameterMap(task, atomicTask);
        Long subDataItemId = Long.valueOf(String.valueOf(param.get("subDataItemId"))) ;
        if (task.getIsODR()) {
            taskDAO.addTaskData(task.getTaskNo(), atomicTask.getAtomicTaskNo(), Convert.toStr(subDataItemId), message, 600l, Boolean.FALSE);
        }
        // Vending接口失败结果处理
        if (StrUtil.equals(TaskType.WsMdmSetParameter.toString(), task.getTaskType())) {
            handleSetParameterResult(task, atomicTask, false, message);
        }
        if (StrUtil.equalsIgnoreCase(task.getTaskType(),TaskType.SetRelayControlMode.name())) {
            assetManagementRemoteService.updateLogRelayControl(task.getTaskNo(), task.getMeterId(), ConfigScheduleTask.ActionType.SET_DEVICE_CONFIG.getType(), SubTaskStatus.Failed.getCode(),null, message);
        }
        return -1;
    }

    private String getAtomicTaskValue(AtomicTask atomicTask, String key) {
        Map<String, Object> map = JSONUtil.toBean(atomicTask.getAtomicTaskParam(), Map.class);
        if (map.containsKey(key) && null != map.get(key)) {
            return (String) map.get(key);
        }
        return "";
    }

    /**
     * 处理Vending Token webservice 接口
     * @param task
     * @param atomicTask
     * @param isSuccess
     * @param failedMessage
     */
    private void handleSetParameterResult(Task task, AtomicTask atomicTask,  boolean isSuccess, String failedMessage) {
        if (isSuccess) {
            // 下发可能多个参数 最后一个都成功的话返回给vending 如果有失败的结果,则马上返回错误并结束后面的参数下发
            if (isSubTaskGroupLastAtomicTask(task, atomicTask)) {
                // send to adapter
                Map<String, Object> taskMap = JSONUtil.toBean(task.getParamJson(), Map.class);
                JSONObject obj = JSONUtil.createObj();
                String vendingXmlStr = (String) taskMap.get("vendingXmlRequest");
                obj.set("VendingType", "setParameter");
                obj.set("vendingXmlRequest", vendingXmlStr);
                obj.set("replyCode", "0.0");
                kafkaTemplate.send(TaskKafkaTopic.TASK_VENDING_RESULT_TO_ADAPTER, JSONUtil.toJsonStr(obj).getBytes(StandardCharsets.UTF_8));
            }
        } else {
            String errorMessage = "";
            if (!isSubTaskGroupLastAtomicTask(task, atomicTask)) {
                removeSetAtomicTaskIfExists(task, atomicTask);
            }
            if (StrUtil.isNotEmpty(failedMessage)) {
                errorMessage = String.format("The item [%s] error: [%s]", getAtomicTaskValue(atomicTask, "refValue"), failedMessage);
            } else {
                errorMessage = String.format("The item [%s] failed", getAtomicTaskValue(atomicTask, "refValue") );
            }
            // send to adapter
            Map<String, Object> taskMap = JSONUtil.toBean(task.getParamJson(), Map.class);
            JSONObject obj = JSONUtil.createObj();
            String tokenXmlStr = (String) taskMap.get("vendingXmlRequest");
            obj.set("VendingType", "setParameter");
            obj.set("vendingXmlRequest", tokenXmlStr);
            obj.set("errorMessage", errorMessage);
            obj.set("replyCode", "0.2");
            kafkaTemplate.send(TaskKafkaTopic.TASK_VENDING_RESULT_TO_ADAPTER, JSONUtil.toJsonStr(obj).getBytes(StandardCharsets.UTF_8));
        }
    }

    private void removeSetAtomicTaskIfExists(Task task, AtomicTask atomicTask) {
        if (StrUtil.equalsIgnoreCase(TaskType.WsMdmSetParameter.toString(), task.getTaskType()) && atomicTask.getGroupId().longValue() == OperationGroup.GeneralParametersSet.getId()) {
            //失败的话 结束后续所有任务
            AtomicTask disconnectTask = taskDAO.atomicTaskQueuePopReverse(task.getTaskNo());
            String termQueueId = taskDAO.getTerminalTaskListKey(task);
            taskDAO.terminalTaskQueuePop(termQueueId);
            taskDAO.atomicTaskQueueDelete(task.getTaskNo());
            taskDAO.clearTaskProcessingAtomic(task.getTaskNo());
            log.info("remove harConnect atomicTask {} ", disconnectTask.getAtomicTaskNo());
        }
    }

}
